package com.ruoyi.mq.config;

import com.rabbitmq.client.Channel;
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.core.*;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.amqp.rabbit.config.SimpleRabbitListenerContainerFactory;
import org.springframework.amqp.rabbit.connection.CachingConnectionFactory;
import org.springframework.amqp.rabbit.connection.ConnectionFactory;
import org.springframework.amqp.rabbit.core.RabbitAdmin;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer;
import org.springframework.amqp.rabbit.listener.api.ChannelAwareMessageListener;
import org.springframework.amqp.rabbit.support.DefaultMessagePropertiesConverter;
import org.springframework.amqp.rabbit.support.MessagePropertiesConverter;
import org.springframework.amqp.support.ConsumerTagStrategy;
import org.springframework.amqp.support.converter.Jackson2JsonMessageConverter;
import org.springframework.amqp.support.converter.MessageConverter;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.ComponentScan;
import org.springframework.context.annotation.Configuration;
import org.springframework.scheduling.concurrent.CustomizableThreadFactory;
import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;

import java.io.IOException;
import java.lang.management.ManagementFactory;
import java.net.InetAddress;
import java.net.ServerSocket;
import java.util.UUID;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.Executor;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;

/**
 * @author jierui
 * @package com.ruoyi.mq.config
 * @date 2021/7/2-9:38
 * @description: rabbit config common
 */
@Configuration
//@ComponentScan({"com.cp.spring.*"})
@Slf4j
public class RabbitMQConfig {

    //队列大小
    public static final int BLOCK_QUEUE = 32 * 1024;


    public static final int INIT_POOL_SIZE=Runtime.getRuntime().availableProcessors();
    @Bean("rabbitExecutor")
    public Executor rabbitExecutor(){

        ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor();
        //核心线程数100 超过100个消费者监听需要修改核心线程数，消费者多线程消费 建议类似1-3 这样配置，这样初始线程池数量为消费者监听类数量总和即可
        //初始线程池数量不能低于concurrency总数。
        executor.setCorePoolSize(100);
        //最大消费者线程数 使用@RabbitListener 注解多线程调用时注意配置线程数总和<最大线程数
        executor.setMaxPoolSize(200);
        //缓冲队列
        executor.setQueueCapacity(BLOCK_QUEUE);
        //允许线程的空闲时间15秒
        executor.setKeepAliveSeconds(15);
        //线程名称
        executor.setThreadNamePrefix("rabbit-exec-pool-");


        //线程池对拒绝任务的处理策略
        //一个被拒绝任务的处理程序，它会丢弃最老的未处理任务
        //executor.setRejectedExecutionHandler(new ThreadPoolExecutor.DiscardOldestPolicy());
        //初始化
        executor.initialize();
        return executor;
    }



    @Bean("rabbitConnExecutor")
    public Executor connectionFactoryThreadPoolExecutor(){

        log.info("当前rabbit-conn-pool核心数*2：{}",INIT_POOL_SIZE);

        ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor();
        //核心线程数5
        executor.setCorePoolSize(INIT_POOL_SIZE*2);
        //最大AMQP Connection size 与最大消费者线程数保持一致
        executor.setMaxPoolSize(200);
        //缓冲队列
        executor.setQueueCapacity(BLOCK_QUEUE);
        //允许线程的空闲时间15秒
        executor.setKeepAliveSeconds(1);
        //线程名称
        executor.setThreadNamePrefix("rabbit-conn-pool-");


        //线程池对拒绝任务的处理策略
        //一个被拒绝任务的处理程序，它会丢弃最老的未处理任务
        //executor.setRejectedExecutionHandler(new ThreadPoolExecutor.DiscardOldestPolicy());
        //初始化
        executor.initialize();
        return executor;
    }


    @Bean
    public ConnectionFactory connectionFactory(RabbitProperties rabbitProperties){
        CachingConnectionFactory connectionFactory = new CachingConnectionFactory();
        connectionFactory.setAddresses(rabbitProperties.getHost());
        connectionFactory.setUsername(rabbitProperties.getUsername());
        connectionFactory.setPassword(rabbitProperties.getPassword());
        connectionFactory.setVirtualHost(rabbitProperties.getVhosts());
        connectionFactory.setExecutor(connectionFactoryThreadPoolExecutor());
        return connectionFactory;
    }

    @Bean
    public SimpleRabbitListenerContainerFactory simpleRabbitListenerContainerFactory(ConnectionFactory connectionFactory){
        SimpleRabbitListenerContainerFactory factory = new SimpleRabbitListenerContainerFactory();
        factory.setConnectionFactory(connectionFactory);
        //手动确认消息
        factory.setAcknowledgeMode(AcknowledgeMode.MANUAL);
        factory.setMessageConverter(messageConverter());
        factory.setTaskExecutor(rabbitExecutor());
        factory.setConsumerTagStrategy(new ConsumerTagStrategy() {
            @Override
            public String createConsumerTag(String queue) {

                ServerSocket serverSocket = null; //读取空闲的可用端口
                String localIp = null;
                try {
                    serverSocket = new ServerSocket(0);
                    localIp = InetAddress.getLocalHost().getHostAddress();
                } catch (IOException e) {
                    e.printStackTrace();
                }
                int port = serverSocket.getLocalPort();

                log.info("rabbit-consumer-tag:{}:{}",localIp,+port);


                //每个消费端都有自己独立的标签
                StringBuilder sb=new StringBuilder();
                sb.append(queue).append("-")
                        .append(localIp).append(":").append(port)
                        .append("-").append(Thread.currentThread().getName());

                return sb.toString();
            }
        });

        return factory;
    }




    //形参名称要与bean的方法名保持一致
    @Bean
    public RabbitAdmin rabbitAdmin(ConnectionFactory connectionFactory) {
        RabbitAdmin rabbitAdmin = new RabbitAdmin(connectionFactory);
        rabbitAdmin.setAutoStartup(true);
        return rabbitAdmin;
    }


    @Bean
    public RabbitTemplate rabbitTemplate(ConnectionFactory connectionFactory) {
        RabbitTemplate rabbitTemplate = new RabbitTemplate(connectionFactory);
        // 设置生产者消息确认
        rabbitTemplate.setConfirmCallback(new RabbitConfirmCallback());
        rabbitTemplate.setReturnCallback(new RabbitConfirmReturnCallBack());
        //消息格式适配
        rabbitTemplate.setMessageConverter(messageConverter());
        rabbitTemplate.setMessagePropertiesConverter(messagePropertiesConverter());
        return rabbitTemplate;
    }


    /**
     * 消息格式转换器
     * @return
     */
    @Bean
    public MessageConverter messageConverter() {
        return new Jackson2JsonMessageConverter();
    }

    @Bean
    public MessagePropertiesConverter messagePropertiesConverter() {
        return new DefaultMessagePropertiesConverter();
    }

}
